<?php
include_once(dirname(__FILE__) . "/../../route/default.php");
include_once(dirname(__FILE__) . "/../config.php");
include_once(dirname(__FILE__) . "/server.php");
include_once(dirname(__FILE__) . "/client.php");


class AnnaClient extends Client {
    const TYPE_UNKNOWN = 0;
    const TYPE_PARSER = 1;
    const TYPE_TRIGGER = 2;

    private $mTitle = null;
    private $mName = null;
    private $mInfo = null;
    private $mType = self::TYPE_UNKNOWN;
    private $mParserId = 0;
    private $mWorking = false;

    public function AnnaClient($server, $fd, $ip, $port) {
        parent::Client($server, $fd, $ip, $port);
    }

    public function onCommand($op, $data) {
        switch ($op) {
        case "register":
            return $this->onRegister($data);
        case "task":
            return $this->onTask($data);
        case "sync":
            return $this->onSync($data);
        }
    }

    private function onRegister($data) {
        $this->mTitle = $data["title"];
        $this->mName = $data["name"];
        $this->mInfo = $data["info"];
        $this->mType = self::TYPE_PARSER;
        $this->log("A parser registered for {$this->mName}.");

        $this->mParserId = 0;
        $parsers = Parser::all();
        foreach ($parsers as $id => $parser) {
            if ($parser->name() == $this->mName) {
                $this->mParserId = $id;
            }
        }
        if ($this->mParserId == 0) {
            $parser = new Parser();
            $parser->setName($this->mName);
            $parser->setTitle($this->mTitle);
            $parser->setInfo($this->mInfo);
            $ret = $parser->save();
            if ($ret !== false) {
                $this->mParserId = $ret;
                $this->log("Add new parser for {$this->mName}.");
            }
        }
    }

    public function onTask($data) {
        $this->mWorking = false;
        $id = $data["id"];
        $this->server()->updateResult($id);
    }

    public function toString() {
        return json_encode(array("fd" => $this->fd(), "addr" => $this->addr(), "type" => $this->mType, "name" => $this->name(), "title" => $this->title(), "working" => $this->isWorking() ? "true" : "false"));
    }

    private function onSync() {
        $this->server()->sync();
    }

    public function isParser() {
        return ($this->mType == self::TYPE_PARSER);
    }

    public function name() {
        return $this->mName;
    }

    public function title() {
        return $this->mTitle;
    }

    public function parserId() {
        return $this->mParserId;
    }

    public function isWorking() {
        return $this->mWorking;
    }

    public function parse($task) {
        $this->mWorking = true;
        $this->send("task", array("id" => $task->id(), "file" => $task->targetUrl(), "upload" => HOME_URL . "?action=api.v1.task.result", "fail" => HOME_URL . "?action=api.v1.task.fail"));
    }
};


class AnnaServer extends Server {
    const PENDING_TIMEOUT = 10000;

    private $clients = array();
    private $mPendingTasks = array();

    protected function onClientConnect($fd, $ip, $port) {
        if (!isset($this->clients[$fd])) {
            $this->clients[$fd] = new AnnaClient($this, $fd, $ip, $port);
        }
        logging::d("Anna", "clients count: " . count($this->clients));
    }

    protected function onCommand($fd, $op, $data) {
        if (!isset($this->clients[$fd])) {
            return;
        }
        logging::d("Anna", "onCommand from [$fd]: [$op] " . print_r($data, true));
        $this->clients[$fd]->onCommand($op, $data);
    }

    public function onInit() {
    }

    protected function onClientClose($fd) {
        if (!isset($this->clients[$fd])) {
            return;
        }
        unset ($this->clients[$fd]);
    }

    private function syncTasks() {
        $tasks = Task::all();
        foreach ($tasks as $id => $task) {
            if (!$task->isOpen()) {
                if (isset($this->mPendingTasks[$id])) {
                    unset($this->mPendingTasks[$id]);
                }
                continue;
            }
            if (!isset($this->mPendingTasks[$task->id()])) {
                $this->mPendingTasks[$task->id()] = $task;
            }
        }
        logging::d("Anna", "pending task count: " . count($this->mPendingTasks));
        foreach ($this->mPendingTasks as $task) {
            logging::d("Anna", "    pending task: " . $task->toString());
        }
    }

    private function openParsers() {
        $parsers = array();
        foreach ($this->clients as $client) {
            if (!$client->isParser()) {
                continue;
            }
            $pid = $client->parserId();
            if (!isset($parsers[$pid])) {
                $parsers[$pid] = array();
            }
            $parsers[$pid][] = $client;
        }
        return $parsers;
    }

    private function parse() {
        $openparsers = $this->openParsers();
        // logging::d("Anna", $openparsers);

        $count = 0;
        foreach ($openparsers as $pid => $parsers) {
            $count += count($parsers);
            logging::i("Anna", "parsers for $pid: " . count($parsers));
            $index = 0;
            foreach ($parsers as $parser) {
                logging::i("Anna", "    [$index] " . $parser->toString());
                $index++;
            }
        }
        logging::i("Anna", "all parsers: " . $count);


        $now = time(null);
        foreach ($this->mPendingTasks as &$task) {
            $pid = $task->parser();
            if (!isset($openparsers[$pid])) {
                continue;
            }
            // if ($task->isPending() && $task->pendingTime() + self::PENDING_TIMEOUT > $now) {
            //     continue;
            // }
            $parsers = $openparsers[$pid];
            foreach ($parsers as $parser) {
                if ($parser->isWorking()) {
                    continue;
                }
                $task->setPending();
                $parser->parse($task);
                logging::d("Anna", "dispatch task {$task->id()}({$task->title()}) to client {$parser->fd()}({$parser->addr()}).");
                break;
            }
        }
    }

    public function updateResult($id) {
        if (isset($this->mPendingTasks[$id])) {
            unset($this->mPendingTasks[$id]);
        }
        $this->sync();
    }

    public function broadcast($op, $data) {
        foreach ($this->clients as $fd => $client) {
            $this->send($fd, $op, $data);
        }
    }

    public function sync() {
        $this->syncTasks();
        $this->parse();
    }

    public function onTick($arg) {
        logging::d("Anna", "AnnaServerTick");
        $this->sync();
    }

    protected function onFirstRef() {
        swoole_timer_tick(10 * 60 * 1000, array($this, "onTick"), null);
    }
};

logging::set_file_prefix("anna-");
logging::set_logging_dir(dirname(__FILE__) . "/../logs/");

$s = new AnnaServer();
$s->start(10123);



